import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

@Slf4j
public class tset {
    @Test
    public void test() {
        //Flux 是 Reactor 框架的一个主要类，用于表示包含零到多个元素的异步序列。
        //创建了一个包含字符串 "Hello", "World", "!" 的 Flux 实例
        Flux<String> stringFlux = Flux.just("Hello", "World", "!");
        Flux<Integer> intFlux = Flux.just(1, 2, 3, 4, 5);

        //订阅了这个 Flux，通过 onNext, onError, onComplete 回调处理每个元素、错误和完成信号
        //方法被调用，注册了三个回调函数
        stringFlux.subscribe(
                System.out::println,// onNext callback (处理每个元素)
                Throwable::printStackTrace,  // onError callback (处理错误)
                () -> System.out.println("Flux completed")  // onComplete callback (处理完成信号)
        );

        //1. map 操作符将每个元素转换为大写
        stringFlux.map(String::toUpperCase)
                .subscribe(System.out::println);

        //2.filter 过滤出偶数
        intFlux.filter(i -> i % 2 == 0)
                .subscribe(System.out::println);

        //3.take 只取前两个元素 take(取出的元素个数)
        intFlux.take(2)
                .subscribe(System.out::println);

        Flux<Integer> flux1 = Flux.just(1, 2, 3);
        Flux<Integer> flux2 = Flux.just(10, 20, 30, 40);

        //4.zip 将从每个上游接收唯一值，并将该值返回给下游信号
        Flux.zip(flux1, flux2, (x, y) -> x + y)
                .subscribe(System.out::println);  // 输出：11, 22, 33
    }

    //5.错误处理
    @Test
    public void errorHandling() {
        Flux<Integer> errorFlux = Flux.concat(
                Flux.just(1, 2, 3),
                Flux.error(new RuntimeException("Error occurred!")),
                Flux.just(4, 5)
        );

        errorFlux
                .onErrorResume(e -> {
                    System.out.println("Handling error: " + e.getMessage());
                    return Flux.just(-1, -2);  // 提供备用数据流
                })
                .subscribe(System.out::println);  // 输出：1, 2, 3, Handling error: Error occurred!, -1, -2

    }

    //6.背压
    @Test
    public void backPressure() throws InterruptedException {
        Flux.range(1, 1000)
                .publishOn(Schedulers.parallel())
                // 使用并行调度器以并行方式生成数据
                .subscribe(data -> {
                    try {
                        Thread.sleep(100);  // 模拟消费者处理每个数据项所需的时间
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("Consumed: " + data);
                });

        Thread.sleep(10000);  // 等待足够长的时间以观察输出
    }

    //7. 线程调度
    @Test
    public void threadScheduling() {
        Flux<Integer> flux = Flux.range(1, 10);
        flux
                .publishOn(Schedulers.elastic())  // 在弹性线程池中执行
                .map(i -> {
                    System.out.println("Mapping " + i + " on thread " + Thread.currentThread().getName());
                    return i;
                })
                .subscribeOn(Schedulers.parallel())  // 在并行线程池中执行
                .subscribe(i -> System.out.println("Received " + i + " on thread " + Thread.currentThread().getName()));
    }

    //Mono 是 Reactor 中表示包含零或一个元素的异步序列的主要类。
    // 它可以发出零个或一个元素，并在结束时发出一个完成信号或错误信号。
    @Test
    public void testMono() {
        Mono<String> mono = Mono.just("Hello, World!");
        mono.subscribe(System.out::println);
    }
}
